/*
* Copyright (C) 2020 ~ 2024 chaigec All Rights Reserved.
*
* Author:     chaigec <chaigec@163.com>
* Maintainer: chaigec <chaigec@163.com>

 * Use of this source code is governed by MIT license that can be found in the
 * LICENSE file in the root of the source tree. All contributing project authors
 * may be found in the AUTHORS file in the root of the source tree.
*/

#include "MkRtmpSession.h"
#include "MkRtmpMuxer.h"
#include "MkCoreConfig.h"
#include "MkRtmpServer.h"
#include "MkRtmpPublisherSource.h"
#include "MkRtmpConumer.h"
#include "MkAmf/MkAmf.h"
#include "MkFlv/MkFlvMuxer.h"
#include "MkFlv/MkFlvDemuxer.h"
#include "MkUtil/MkBuffer.h"
#include "MkEvent/MkEvent.h"
#include "MkUtil/MkMd5.h"
#include "MkUtil/MkUtil.h"
#include "MkUtil/MkLog.h"

#ifndef _WIN32
static int ClkTck = 0;
#include <sys/times.h>
#endif
#define RtmpOutChunkSize  60000
CMkRtmpSession::CMkRtmpSession(CMkEvent* pEvent, CMkTimer* pTimer, CMkTcpServer* pServer, const MkMediaCallbacks& MediaCallbacks)
    : CMkTcpSession(pEvent, pTimer, pServer)
    , m_RtmpState(RtmpStateInit)
    , m_pRtmpPublishSource(nullptr)
    , m_pRtmpConsumer(nullptr)
    , m_MediaCbs(MediaCallbacks)
{
    Int8 pBuf[RtmpSigSize] = { 0 };
    Uint32 NowTime;
#ifdef _WIN32
    NowTime = timeGetTime();
#else
    struct tms t;
    if (0 == ClkTck) {
        ClkTck = sysconf(_SC_CLK_TCK);
    }
    NowTime = times(&t) * 1000 / ClkTck;
#endif
    pBuf[0] = (NowTime >> 24) & 0xFF;
    pBuf[1] = (NowTime >> 16) & 0xFF;
    pBuf[2] = (NowTime >> 8) & 0xFF;
    pBuf[3] = NowTime & 0xFF;
    for (int i = 8; i < RtmpSigSize; i++) {
        pBuf[i] = rand() & 0xFF;
    }
    m_LocalSig = MkString(pBuf, RtmpSigSize);
}

Uint32 CMkRtmpSession::SendFlvStream(const MkEsPacket& EsPacket, const CMkBufferList& BufferList)
{
    Uint32 ErrorCode = NoneError;
    //has un send buffer big than half 1MB
    if ((GetUnSendBufferLength() > LenHalfM && EsPacket.bKeyPacket) || !m_ListCacheStream.empty()) {
        MkList<MkCacheStreamPacket>::iterator it;
        if (EsPacket.bKeyPacket) {
            it = m_ListCacheStream.begin();
            for (; it != m_ListCacheStream.end(); it++) {
                it->CacheBufferList.Clear();
            }
            m_ListCacheStream.clear();
        }
        MkCacheStreamPacket CacheStreamPacket;
        CMkMediaUtil::WriteCacheStream(EsPacket, BufferList, CacheStreamPacket);
        m_ListCacheStream.push_back(CacheStreamPacket);
        //send all cache stream packet
        if (GetUnSendBufferLength() < Len20K) {
            it = m_ListCacheStream.begin();
            for (; it != m_ListCacheStream.end(); it++) {
                CMkBufferList TmpBufferList;
                it->CacheBufferList.GetBufferList(TmpBufferList);
                MkRtmpCommandType CmdType = RtmpMsgAmf0;
                if (it->CodecType > MkCodecCavs) {
                    CmdType = RtmpMsgAudio;
                } else if (it->CodecType > MkCodecUnknown) {
                    CmdType = RtmpMsgVideo;
                }
                CMkBufferList SendList;
                m_nRecvOffset = 0;
                Uint32 ErrorCode = CMkRtmpMuxer::PacketRtmpPacket(m_OutChunkSize, MkRtmpHeaderFmt0, CmdType, ChunkSystem, 0, m_pPacketBuffer, m_PacketMaxLen, m_nRecvOffset, TmpBufferList, SendList);
                ErrorCode = Send(SendList);
                it->CacheBufferList.Clear();
            }
            m_ListCacheStream.clear();
        }
    } else {
        MkRtmpCommandType CmdType = RtmpMsgAmf0;
        if (EsPacket.CodecType > MkCodecCavs) {
            CmdType = RtmpMsgAudio;
        } else if (EsPacket.CodecType > MkCodecUnknown) {
            CmdType = RtmpMsgVideo;
        }
        CMkBufferList SendList;
        m_nRecvOffset = 0;
        Uint32 ErrorCode = CMkRtmpMuxer::PacketRtmpPacket(m_OutChunkSize, MkRtmpHeaderFmt0, CmdType, ChunkSystem, 0, m_pPacketBuffer, m_PacketMaxLen, m_nRecvOffset, BufferList, SendList);
        MkCheckErrorLog(ErrorCode, "packet rtmp packet failed\n");
        ErrorCode = Send(SendList);
    }
    return ErrorCode;
}

Uint32 CMkRtmpSession::OnRecv()
{
    Uint32 nToRead = m_RecvMaxLen - m_nRecvOffset;
    if (0 == nToRead) {
        MkErrorLog("buffer size:%d not parse success\n", m_RecvMaxLen);
        Close();
        return UnknownError;
    }
    Uint32 ErrorCode = Recv(m_pRecvBuffer + m_nRecvOffset, nToRead);
    MkCheckErrorLog(ErrorCode, "rtmp session recv failed\n");
    m_nRecvOffset += nToRead;
    Uint32 nUsed = 0;
    while (nUsed + 12 <= m_nRecvOffset) {
        Uint32 CurUsed = 0;
        switch (m_RtmpState) {
        case RtmpStateInit:
            ErrorCode = ParseHandSharkeC0(m_pRecvBuffer + nUsed, m_nRecvOffset - nUsed, CurUsed);
            break;
        case RtmpStateC0:
            ErrorCode = ParseHandSharkeC1(m_pRecvBuffer + nUsed, m_nRecvOffset - nUsed, CurUsed);
            SendHandSharkeS0S1S2();
            break;
        case RtmpStateS2:
            ErrorCode = ParseHandSharkeC2(m_pRecvBuffer + nUsed, m_nRecvOffset - nUsed, CurUsed);
            break;
        default:
            ErrorCode = ParseRtmpMessage(m_pRecvBuffer + nUsed, m_nRecvOffset - nUsed, CurUsed);
            break;
        }
        if (NoneError == ErrorCode) {
            nUsed += CurUsed;
        } else if (NotHasEnoughBuffer == ErrorCode) {
            break;
        } else {
            MkErrorLog("recv error msg close it\n");
            Close();
            return SocketClosed;
        }
    }
    if (nUsed < m_nRecvOffset
        && nUsed>0) {
        memmove(m_pRecvBuffer, m_pRecvBuffer + nUsed, m_nRecvOffset - nUsed);
    }
    m_nRecvOffset -= nUsed;
    return NoneError;
}

Uint32 CMkRtmpSession::OnEventRemoved()
{
    if (m_pRtmpConsumer) {
        m_pRtmpConsumer->NoneRtmpSession();
        if (m_MediaCbs.ConsumerClosedCb) {
            m_MediaCbs.ConsumerClosedCb(m_pRtmpConsumer->GetParentSessionId(), MkMediaConsumeRtmp, m_pRtmpConsumer->GetConsumerId());
        }
    }
    if (m_pRtmpPublishSource) {
        m_pRtmpPublishSource->NoneRtmpSession();
        if (m_MediaCbs.SourceDeleteCb) {
            m_MediaCbs.SourceDeleteCb(m_pRtmpPublishSource->GetSession());
        }
    }
    return  CMkTcpSession::OnEventRemoved();
}

Uint32 CMkRtmpSession::OnCommandRequest(const MkString& Command, const MkVector<MkAmfObject>& VecObj)
{
    //MkDebugLog("recv cmd:%s\n", Command.c_str());
    if ("@setDataFrame" == Command) {
        return OnSetDataFrame(VecObj);
    }
    Uint32 ErrorCode = NoneError;
    MkVector<MkAmfObject>::const_iterator it = VecObj.begin();
    if (it == VecObj.end()
        || MkAmfNumber != it->DataType) {
        MkErrorLog("recv invalid command\n");
        return InvalidParams;
    }
    Uint32 TransactionId = static_cast<Uint32>(it->NumberValue);
    it++;
    if ("connect" == Command) {
        if (MkAmfTypeObject != it->DataType) {
            MkErrorLog("recv invalid connect command message\n");
            return InvalidParams;
        }
        ErrorCode = SendChunkSize(RtmpOutChunkSize);
        MkCheckErrorLog(ErrorCode, "set rtmp chunk size failed\n");
        ErrorCode = SendWindownAcknowledgementSize();
        MkCheckErrorLog(ErrorCode, "set acknowledgement failed\n");
        ErrorCode = SendSetPeerBandWidth(600000);
        MkCheckErrorLog(ErrorCode, "set set peer band width failed\n");
        ErrorCode = SendConnectResult(TransactionId, it->VecObjectValue);
        MkCheckErrorLog(ErrorCode, "send connect result failed\n");
        ErrorCode = SendOnBWDone();
    } else if ("createStream" == Command) {
        ErrorCode = SendCreateStreamResult(TransactionId);
    } else if ("publish" == Command) {
        if (it == VecObj.end()) {
            MkErrorLog("recv invalid publish command message\n");
            return InvalidParams;
        }
        it++;
        if (it == VecObj.end() || MkAmfString != it->DataType) {
            MkErrorLog("recv invalid publish command message\n");
            return InvalidParams;
        }
        m_StreamPath = it->strValue;
        it++;
        if (it == VecObj.end() || MkAmfString != it->DataType) {
            MkErrorLog("recv invalid publish command message\n");
            return InvalidParams;
        }
        m_AppName = it->strValue;
        ErrorCode = ParsePublish(TransactionId);
    } else if ("play" == Command) {
        if (it == VecObj.end()) {
            MkErrorLog("recv invalid play command message\n");
            return InvalidParams;
        }
        it++;
        if (it == VecObj.end() || MkAmfString != it->DataType) {
            MkErrorLog("recv invalid play command message\n");
            return InvalidParams;
        }
        m_StreamPath = it->strValue;
        ErrorCode = ParsePlay(TransactionId);
    } else if ("pause" == Command) {
        if (it == VecObj.end()) {
            return InvalidParams;
        }
        it++;
        if (it == VecObj.end() || MkAmfBoolean != it->DataType) {
            return InvalidParams;
        }
        BOOL bPause = it->bValue;
        it++;
        if (it == VecObj.end() || MkAmfNumber != it->DataType) {
        }
        Uint32 MilliSec = static_cast<Uint32>(it->NumberValue);
        ErrorCode = ParsePause(TransactionId, bPause, MilliSec);
    } else if ("seek" == Command) {
        if (it == VecObj.end()) {
            return InvalidParams;
        }
        it++;
        if (it == VecObj.end() || MkAmfNumber != it->DataType) {
        }
        Uint32 MilliSec = static_cast<Uint32>(it->NumberValue);
        ErrorCode = ParseSeek(TransactionId, MilliSec);
    }
    return ErrorCode;
}

Uint32 CMkRtmpSession::OnVideo(const Uint8* pBuf, Uint32 nLen, Uint32 TimeStamp)
{
    if (m_pRtmpPublishSource) {
        m_pRtmpPublishSource->InputVideo(pBuf, nLen, TimeStamp);
    }
    return NoneError;
}

Uint32 CMkRtmpSession::OnAudio(const Uint8* pBuf, Uint32 nLen, Uint32 TimeStamp)
{
    if (m_pRtmpPublishSource) {
        m_pRtmpPublishSource->InputAudio(pBuf, nLen, TimeStamp);
    }
    return NoneError;
}

Uint32 CMkRtmpSession::OnSetDataFrame(const MkVector<MkAmfObject>& VecObj)
{
    if (m_pRtmpPublishSource) {
        m_pRtmpPublishSource->InputSetDataFrame(VecObj);
        m_pRtmpPublishSource->m_ResultCb(m_pRtmpPublishSource, NoneError);
    }
    return NoneError;
}

Uint32 CMkRtmpSession::ParseHandSharkeC0(const Uint8* pBuf, Uint32 nLen, Uint32& nUsed)
{
    if (nLen < 1) {
        return NotHasEnoughBuffer;
    }
    if (RtmpVersion != *pBuf) {
        MkErrorLog("rtmp version:%d error\n", *pBuf);
        return InvalidParams;
    }
    nUsed = 1;
    m_RtmpState = RtmpStateC0;
    return NoneError;
}

Uint32 CMkRtmpSession::ParseHandSharkeC1(const Uint8* pBuf, Uint32 nLen, Uint32& nUsed)
{
    if (nLen < RtmpSigSize) {
        return NotHasEnoughBuffer;
    }
    m_RemoteSig = MkString((Int8*)pBuf, RtmpSigSize);
    nUsed = RtmpSigSize;
    m_RtmpState = RtmpStateC1;
    return NoneError;
}

Uint32 CMkRtmpSession::ParseHandSharkeC2(const Uint8* pBuf, Uint32 nLen, Uint32& nUsed)
{
    if (nLen < RtmpSigSize) {
        return NotHasEnoughBuffer;
    }
    if (memcmp(pBuf, m_LocalSig.c_str(), m_LocalSig.size())) {
        MkErrorLog("local sig not recv sig\n");
        return InvalidParams;
    }
    nUsed = RtmpSigSize;
    m_RtmpState = RtmpStateC2;
    return NoneError;
}

Uint32 CMkRtmpSession::ParsePublish(const Uint32& TransactionId)
{
    MkString MediaPath = "/" + m_AppName + "/" + m_StreamPath;
    if (!m_MediaCbs.SourceAddCb
        || !m_MediaCbs.GetSourceParamCb) {
        return SystemError;
    }
    CMkVideoParam VideoParam;
    CMkAudioParam AudioParam;
    Uint32 Duration;
    Uint32 FileSize;
    MkString SourceSession;
    Uint32 ErrorCode = m_MediaCbs.GetSourceParamCb(MediaPath, VideoParam, AudioParam, Duration, FileSize, SourceSession);
    if (NoneError == ErrorCode) {
        return SendPublishResult(TransactionId, FALSE);
    }
    m_pRtmpPublishSource = new CMkRtmpPublisherSource(this);
    m_pRtmpPublishSource->SetAppName(m_AppName);
    m_pRtmpPublishSource->SetStreamPath(m_StreamPath);
    m_pRtmpPublishSource->StartMediaSource([this, MediaPath](CMkMediaSource* pMediaSource, Uint32 ErrorCode) {
        if (NoneError == ErrorCode) {
            MkString SourceSession = CMkMd5Encode::Encode(MediaPath);
            pMediaSource->SetSession(SourceSession);
            m_MediaCbs.SourceAddCb(pMediaSource);
        }
        return NoneError;
    });
    SendPublishResult(TransactionId, TRUE);
    return NoneError;
}

Uint32 CMkRtmpSession::ParsePlay(const Uint32& TransactionId)
{
    if (!m_MediaCbs.GetSourceParamCb
        || !m_MediaCbs.ConsumerAddCb) {
        return UnSupported;
    }
    MkString MediaPath = "/" + m_AppName + "/" + m_StreamPath;
    CMkVideoParam VideoParam;
    CMkAudioParam AudioParam;
    Uint32 Duration;
    Uint32 FileSize;
    MkString    SourceSession;
    Uint32 ErrorCode = m_MediaCbs.GetSourceParamCb(MediaPath, VideoParam, AudioParam, Duration, FileSize, SourceSession);
    SendPlayResult(TransactionId, NoneError == ErrorCode);
    SendStreamBegin();
    m_pRtmpConsumer = new CMkRtmpConsumer(this);
    m_MediaCbs.ConsumerAddCb(SourceSession, new CMkRtmpConsume, m_pRtmpConsumer);
    return NoneError;
}

Uint32 CMkRtmpSession::ParsePause(const Uint32& TransactionId, BOOL bPause, Uint32 MilliSec)
{
    Uint32 ErrorCode = UnknownError;
    if (!m_MediaCbs.ConsumerPlayCtrlCb
        || !m_pRtmpConsumer) {
        return UnSupported;
    }
    ErrorCode = m_MediaCbs.ConsumerPlayCtrlCb(m_pRtmpConsumer, bPause ? MkControlPause : MkControlRecovery, 1.0f, MilliSec / 1000);
    return SendPauseResult(TransactionId, bPause, NoneError == ErrorCode);
}

Uint32 CMkRtmpSession::ParseSeek(const Uint32& TransactionId, Uint32 MilliSec)
{
    Uint32 ErrorCode = UnknownError;
    if (!m_MediaCbs.ConsumerPlayCtrlCb
        || !m_pRtmpConsumer) {
        return UnSupported;
    }
    ErrorCode = m_MediaCbs.ConsumerPlayCtrlCb(m_pRtmpConsumer, MkControlSeek, 1.0f, MilliSec / 1000);
    return SendSeekResult(TransactionId, NoneError == ErrorCode);
}

Uint32 CMkRtmpSession::SendHandSharkeS0S1S2()
{
    CMkBufferList BufferList;
    Uint8 uRtmpVersion = RtmpVersion;
    BufferList.Append(&uRtmpVersion, 1);
    BufferList.Append(m_LocalSig.c_str(), m_LocalSig.size());
    BufferList.Append(m_RemoteSig.c_str(), m_RemoteSig.size());
    m_RtmpState = RtmpStateS2;
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendChunkSize(Uint32 ChunkSize)
{
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketSetChunkSize(ChunkSize, m_OutChunkSize, m_pPacketBuffer, m_PacketMaxLen, m_nPacketOffset, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect failed\n");
    m_OutChunkSize = ChunkSize;
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendAcknowledgement()
{
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketAcknowledgement(m_SendRecvSize, m_OutChunkSize, m_pPacketBuffer, m_PacketMaxLen, m_nPacketOffset, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendWindownAcknowledgementSize()
{
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketWindownAcknowledgementSize(m_nServerBw, m_OutChunkSize, m_pPacketBuffer, m_PacketMaxLen, m_nPacketOffset, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendSetPeerBandWidth(Uint32 BandWidth)
{
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketPeerBandwidth(BandWidth, 2, m_OutChunkSize, m_pPacketBuffer, m_PacketMaxLen, m_nPacketOffset, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendStreamBegin()
{
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketStreamBegin(1, m_OutChunkSize, m_pPacketBuffer, m_PacketMaxLen, m_nPacketOffset, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendOnBWDone()
{
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    CMkBufferList BufferList;
    Uint32 ErrorCode = CMkRtmpMuxer::PacketCommandMessage(m_OutChunkSize, "onBWDone", 0, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet on bw done failed\n");
    AddMapTransaction(m_TransactionId, "publish");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendConnectResult(const Uint32& TransactionId, const MkVector<MkAmfObject>& VecObj)
{
    if (1 != TransactionId) {
        return InvalidParams;
    }
    MkVector<MkAmfObject>::const_iterator it = VecObj.begin();
    for (; it != VecObj.end(); it++) {
        if ("app" == it->Name) {
            m_AppName = it->strValue;
        }
    }
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecInfoObj;
    VecInfoObj.push_back(MkAmfObject(MkAmfNull));
    MkAmfObject CmdObject(MkAmfTypeObject);
    CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetConnection.Connect.Success"));
    CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "status"));
    CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "Connection succeeded"));
    VecInfoObj.push_back(CmdObject);
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, "_result", 1, VecInfoObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet connect result failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendCreateStreamResult(const Uint32& TransactionId)
{
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    VecObj.push_back(MkAmfObject("", MkAmfNumber, (double)1));
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, "_result", TransactionId, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet create stream result failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendPlayResult(const Uint32& TransactionId, BOOL bSuccess)
{
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    MkAmfObject CmdObject(MkAmfTypeObject);
    if (bSuccess) {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetStream.Play.Start"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "status"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "Start Playing"));
    } else {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetStream.Play.StreamNotFound"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "error"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "can not find stream"));
    }
    VecObj.push_back(CmdObject);
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, "onStatus", TransactionId, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet create stream result failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendPauseResult(const Uint32& TransactionId, BOOL bPause, BOOL bSuccess)
{
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    MkAmfObject CmdObject(MkAmfTypeObject);
    if (bSuccess) {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, bPause ? "NetStream.Pause.Notify" : "NetStream.Unpause.Notify"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "status"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, bPause ? "Pause succeeded" : "Unpause succeeded"));
    }/* else {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, bPause? "NetStream.Pause.Notify":"NetStream.Unpause.Notify"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "error"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, bPause?"Pause failed":"Unpause failed"));
    }*/
    VecObj.push_back(CmdObject);
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, bSuccess ? "_result" : "_error", TransactionId, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet create stream result failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendSeekResult(const Uint32& TransactionId, BOOL bSuccess)
{
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    MkAmfObject CmdObject(MkAmfTypeObject);
    if (bSuccess) {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetStream.seek.Notify"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "status"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "Seek succeeded"));
    }/* else {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, bPause? "NetStream.Pause.Notify":"NetStream.Unpause.Notify"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "error"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, bPause?"Pause failed":"Unpause failed"));
    }*/
    VecObj.push_back(CmdObject);
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, bSuccess ? "_result" : "_error", TransactionId, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet create stream result failed\n");
    return CMkTcpSession::Send(BufferList);
}

Uint32 CMkRtmpSession::SendPublishResult(const Uint32& TransactionId, BOOL bSuccess)
{
    CMkBufferList BufferList;
    MkVector<MkAmfObject> VecObj;
    VecObj.push_back(MkAmfObject(MkAmfNull));
    MkAmfObject CmdObject(MkAmfTypeObject);
    if (bSuccess) {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetStream.Publish.Start"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "status"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "Start Publishing Stream"));
    } else {
        CmdObject.VecObjectValue.push_back(MkAmfObject("code", MkAmfString, "NetStream.Publish.BadName"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("level", MkAmfString, "error"));
        CmdObject.VecObjectValue.push_back(MkAmfObject("description", MkAmfString, "Stream Already Have"));
    }
    VecObj.push_back(CmdObject);
    Uint32 ErrorCode = CMkRtmpMuxer::PacketResponseMessage(m_OutChunkSize, "onStatus", TransactionId, VecObj, m_pPacketBuffer, m_PacketMaxLen, BufferList);
    MkCheckErrorLog(ErrorCode, "packet create stream result failed\n");
    return CMkTcpSession::Send(BufferList);
}

CMkRtmpSession::~CMkRtmpSession()
{
    MkDelete(m_pRtmpPublishSource);
    MkDelete(m_pRtmpConsumer);
}